/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package postmaster;

import MessageHandler.message_type;
import MessageHandler.messagehandler;
import MessageHandler.messages;
import MessageHandler.submessage;
import configuration.config_manager;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.net.SocketException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 *
 * @author maniksin
 */
public class postmaster {
    
    public static final int DEFAULT_BIND_PORT = 5000;
    //private final int MAX_PACKET_SIZE = 100;
    
    public static final String version = "1.19";
    
    public config_manager conf;
    private int bind_port;
    
    public Set peers;      // set of peers connected to postmaster
    
    private boolean postmaster_ready;
    private postmaster_errors error_code;
    
    public DatagramSocket sock;
    
    public postmaster(config_manager conf) {
        // Check if we know our port
        this.conf = conf;
        error_code = postmaster_errors.SUCCESS;
        
        bind_port = conf.get_my_port(DEFAULT_BIND_PORT);
        
        postmaster_ready = true;
        try {
            sock = new DatagramSocket(bind_port);
        } catch (SocketException ex) {
            postmaster_ready = false;
            error_code = postmaster_errors.BIND_FAILED;            
        }
        
        if (postmaster_ready) {
            System.out.println("Postmaster registered successfully on port " + bind_port);
        }
    }
    
    public void close_postmaster() {
        sock.close();
    }
    
    public postmaster_errors is_ready() {
        if (postmaster_ready) {
            return postmaster_errors.SUCCESS;
        } else {
            return error_code;
        }
    }
    
    public void send_message(DatagramPacket dp) {
        try {
            sock.send(dp);
        } catch (IOException ex) {
            Logger.global.log(Level.WARNING, "Failed to send packet out. IOException {0}", ex.getMessage());
        }
        
    }
    public void send_message(DatagramPacket dp, peer p) {
        
        if (p.next_hop != null) {
            // This is not an adjacent peer
            p = p.next_hop;
        }
        
        dp.setAddress(p.get_ip());
        dp.setPort(p.get_port());
        try {
            sock.send(dp);
        } catch (IOException ex) {
            Logger.global.log(Level.WARNING, "Failed to send packet out. IOException {0}", ex.getMessage());
        }
    }
    
    public boolean is_peer_compatible(String version) {
        String[] self_versions = postmaster.version.split("\\.");
        String[] peer_versions = version.split("\\.");
        
        if (self_versions[0].equals(peer_versions[0])) {
            return true;
        }
        return false;
    }
    
    public void report_incompatibility(DatagramPacket dp) {
        
        messagehandler rcvd_msg = messagehandler.decode_message(dp);
        messagehandler msg = new messagehandler();
        
        msg.msg_id = messages.HELLO_PACKET;
        msg.source = conf.my_name;
        msg.dest = rcvd_msg.source;
        msg.sub_msg_id = submessage.NOT_APPLICABLE;
        msg.type = message_type.ACKNOWLEDGE_NONE;
        msg.user_ip = version;
        msg.user_port = 1;  // 1 means do not reply
        
        msg.cmd_string = new String[1];
        msg.cmd_string[0] = "0";
        
        DatagramPacket reply_dp = messagehandler.create_message(msg);
        reply_dp.setAddress(dp.getAddress());
        reply_dp.setPort(dp.getPort());
        send_message(reply_dp);
        
    }
    public void sayHello(peer p) {
        messagehandler msg = new messagehandler();
        
        msg.msg_id = messages.HELLO_PACKET;
        msg.source = conf.my_name;
        msg.dest = p.get_name();
        msg.sub_msg_id = submessage.NOT_APPLICABLE;
        msg.type = message_type.ACKNOWLEDGE_NONE;
        msg.user_ip = version;  // user_ip field used as version in HELLO packet
        msg.user_port = 0;
        
        peer next_peer;
        Iterator i = peers.iterator();
        Set hello_peers = new HashSet();
                
        while (i.hasNext()) {
            next_peer = (peer) i.next();
            if (next_peer != p && next_peer.next_hop != p && next_peer.state != peer_state.PEER_DEAD) {
                hello_peers.add(next_peer);
            }
        }
        
        msg.cmd_string = new String[hello_peers.size() + 1];
        msg.cmd_string[0] = Integer.toString(hello_peers.size());
        
        i = hello_peers.iterator();
        int j = 1;
        while (i.hasNext()) {
            msg.cmd_string[j] = ((peer)i.next()).get_name();
        }
        
        DatagramPacket dp = messagehandler.create_message(msg);
        send_message(dp, p);
        
        Logger.global.log(Level.FINER, "Sending hello to {0}", p.get_name());
    }
    
    public void start_monitoring(peer p) {
        
        long delay = peer_timer.HELLO_INTERVAL;
        p.hello_timer.schedule(new peer_timer(p, this, peer_timer.HELLO_TIMER), delay);
        Logger.global.log(Level.FINE, "Hand-shaking started for {0}", p.get_name());
    }
    
    public boolean hello_received(peer p, DatagramPacket dp) {
        
        
        if (!dp.getAddress().equals(p.get_ip())) {
            //2 different peers with same name!!
            Logger.global.log(Level.WARNING,"Hello received from {0}" + " from a different location."
                    + " Not considered legitimate!!", p.get_name());
            
            // @todo: Notify remote peer of duplicate name, and to shut down
            // Hello packet does not use sub_msg_id. Field can be used to notify
            // the same to the peer
            return false;
        }
        
        //Peer may restart binding to a different port on remote machine
        //So, keep updating the latest port.
        p.set_port(dp.getPort());
        
        p.state = peer_state.PEER_RESPONDING;
        Logger.global.log(Level.FINER, "Hello received from {0}", p.get_name());
        
        messagehandler msg = messagehandler.decode_message(dp);
        int num_connected_peers = Integer.parseInt(msg.cmd_string[0]);
        
        for (int i = 1; i <= num_connected_peers; i++) {
            String peer_name = msg.cmd_string[i];
            peer stored_peer = peer.get_peer_by_name(peer_name, peers);
            
            if (stored_peer != null) {
                // Peer already in database
                stored_peer.state = peer_state.PEER_RESPONDING;
                stored_peer.next_hop = p;       // Override if adjacent peer becomes a connected peer
            } else {
                stored_peer = new peer(peer_name, p, peer_state.PEER_RESPONDING);
                peers.add(stored_peer);
                Logger.global.info("Peer " + peer_name + " added to indirectly connected neighbors");
                
                long delay = peer_timer.NEIGHBOR_DEAD_TIMER;
                stored_peer.neighbor_timer.schedule(new peer_timer(stored_peer, this, peer_timer.NEIGHBOR_DEAD), delay);
            }
        }
        
        return true;        
    }
    
    public void start_postmaster() {
        
        peers = new HashSet();
        
        if (conf.is_client) {
            // Server credentials are present.
            // Add server as a peer, and say hello
            InetAddress ip;
            int port;
            String name;
            ip = conf.server_ip;
            port = conf.server_port;
            name = conf.server_name;
            peer server = new peer(ip, port, name, peer_state.PEER_DEAD);
          
            Logger.global.log(Level.INFO, "Server {0} added permanantly to neighbour's list", server.get_name());
            peers.add(server);
            
            sayHello(server);
            start_monitoring(server);
        }
    
        // Start http router if it is provided
        if (conf.http_router != null) {
            http_router router_thread = new http_router(this);
            router_thread.start();
        }
        
        Logger.global.log(Level.INFO,"Postmaster version " + version + 
                " started listening on socket as {0}...", conf.my_name);
        
        while (true) {
            DatagramPacket dp = new DatagramPacket(
                    new byte[messagehandler.MAX_MSG_PAYLOAD_LEN], messagehandler.MAX_MSG_PAYLOAD_LEN);
            peer stored_peer;
            messagehandler msg;
        
            try {
                sock.receive(dp);
            } catch (IOException ex) {
                 return;
            }    
            
            msg = messagehandler.decode_message(dp);
            stored_peer = messagehandler.find_sender(dp, peers);
            
            
            switch(msg.msg_id) {
                case HELLO_PACKET:
                    
                    // Check for compatibility
                    if (!is_peer_compatible(msg.user_ip)) {
                        
                        Logger.global.log(Level.INFO, "Connecting peer {0} has incompatible version {1}", new Object[]{msg.source, msg.user_ip});
                        
                        if (msg.user_port == 0) {
                            //Peer does not know about incompatibility
                            // Send a hello packet to notify incompatibility
                            report_incompatibility(dp);
                        }
                        
                        // If peer is server, then no need to run client further
                        if (conf.is_client) {
                            if (conf.server_ip.equals(dp.getAddress())) {
                                Logger.global.log(Level.SEVERE, "Server has incompatible version: {0}. Exiting...", msg.user_ip);
                                
                                stored_peer.hello_timer.cancel();
                                stored_peer.neighbor_timer.cancel();
                                peers.remove(stored_peer);
                                return;
                            }
                        }
                        
                        break;
                    }
                    
                    if (stored_peer != null) {
                        // Peer already stored with us.
                        peer_state stored_peer_initial_state = stored_peer.state;
                        if (hello_received(stored_peer, dp) == false) {
                            break;  // Not from a valid peer, no further processing
                        }
                        if (stored_peer_initial_state != peer_state.PEER_DEAD) {
                            Logger.global.log(Level.FINER, "Hello received from: {0}", stored_peer.get_name());
                        } else {
                            
                            Logger.global.log(Level.INFO, "Connected to: {0}", stored_peer.get_name());
                            long delay = peer_timer.NEIGHBOR_DEAD_TIMER;
                            stored_peer.neighbor_timer.schedule(
                                    new peer_timer(stored_peer, this, peer_timer.NEIGHBOR_DEAD), delay);
                            Logger.global.log(Level.FINE, "Monitoring started for peer {0}", stored_peer.get_name());
                        }
                        
                    } else {
                        // New client
                        
                        peer p = new peer(dp, peer_state.PEER_RESPONSE_AWAITED);
                        Logger.global.log(Level.INFO, "Peer {0} added to list of neighbours", p.get_name());
                        peers.add(p);
                        start_monitoring(p);
                        hello_received(p, dp);
                        // Start neighbor_dead timer
                        long delay = peer_timer.NEIGHBOR_DEAD_TIMER;
                        p.neighbor_timer.schedule(new peer_timer(p, this, peer_timer.NEIGHBOR_DEAD), delay);
                    }
                    break;
                
                case PEER_COMMAND:
                case PEER_REPLY:
                    
                    if (msg.dest.equals(conf.my_name) == false) {
                        // Its not for us
                        
                        // First check if source is known
                        if (messagehandler.find_sender(dp, peers) == null) {
                            // Packet from unknown peer
                            Logger.global.log(Level.WARNING, "Message <{0}> received from unknown peer. Not handled", msg.sub_msg_id.cmd);
                            break;
                        }
                        
                        peer p = peer.get_peer_by_name(msg.dest, peers);
                        if (p != null) {
                            Logger.global.log(Level.INFO, "Forwarded <{0}> message to peer", msg.sub_msg_id.cmd);
                            send_message(dp, p);
                        } else {
                            // We don't know the peer
                            Logger.global.log(Level.WARNING, "Peer {0} not known. Message not forwarded", msg.dest);
                        }
                        
                        break;                        
                    }
                    
                    if (stored_peer != null) {
                        Logger.global.log(Level.INFO, "Received <{0}> from {1}.", new Object[]{msg.msg_id.toString(), msg.source});
                        messagehandler.handle_message(dp, this);
                    } else {
                        Logger.global.log(Level.WARNING, "Received <{0}/{1}> is dropped!! Peer not known ({2}.", new Object[]{msg.msg_id, msg.sub_msg_id.cmd, msg.source});
                    }
                    break;
                    
                case USER_COMMAND: 
                    peer p = peer.get_peer_by_dest_name(dp, peers);
                    if (p != null && p.state != peer_state.PEER_DEAD) {
                        Logger.global.log(Level.INFO, "Received a USER command {0} for {1} from {2}/{3}", new Object[]{msg.sub_msg_id.cmd, msg.dest, dp.getAddress().toString(), dp.getPort()});
                        DatagramPacket cmd_dp = messagehandler.create_command(dp, conf);
                        send_message(cmd_dp, peer.get_peer_by_dest_name(dp, peers));
                    } else {
                        if (msg.dest.compareTo(conf.my_name) == 0) {
                            // Message for self
                            messagehandler.handle_message(dp, this);
                        } else {
                            Logger.global.log(Level.WARNING, "Received a USER command for an unknown/lost peer: {0}. Not handled!!", msg.dest);
                        }
                    }
                    break;
                default:
                    Logger.global.log(Level.WARNING, "Unknown command: {0} received. Not handled!!", msg.msg_id);
                    break;
            }          
            
        }
        
        
    }
}
